package streaming

import org.apache.spark.streaming.{Durations, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}

object DSWordCount extends App {
  val sparkConf = new SparkConf()
  sparkConf.setAppName("DSWordCount")
  sparkConf.setMaster("local[4]")
  val sparkContext = new SparkContext(sparkConf)
  /**
   * 5s 一个窗口 即一个批次
   */
  val ssc = new StreamingContext(sparkContext, Durations.seconds(5))
  /**
   * 先在 node1 上启动 nc -lk 9999
   */
  val ds = ssc.socketTextStream("node1", 9999)
  ds.map((_, 1)).reduceByKey(_ + _).print()

  ssc.start()
  ssc.awaitTermination()
  ssc.stop()
}
